/*
 * @(#)DataCenterConsumer.java 2011-11-11
 *
 * Copyright 2011 北龙中网（北京）科技有限责任公司. All rights reserved.
 */
package cn.shangjia.shop.sync;

import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.annotation.Resource;

import cn.shangjia.common.domain.Domain;
import cn.shangjia.common.exceptions.Logger;
import cn.shangjia.common.mq.exec.KnetMessageListener;
import cn.shangjia.common.mq.exec.message.KnetMessage;
import cn.shangjia.common.mq.exec.message.KnetMessageHead;
import cn.shangjia.common.mq.message.MessageBean;
import cn.shangjia.common.util.StringUtils;
import cn.shangjia.shop.log.ShopLoggerFactory;
import cn.shangjia.shop.service.MqLogService;
import cn.shangjia.shop.sync.process.ProcessTask;
import cn.shangjia.shop.sync.service.SyncService;

/**
 * 类说明.
 * 
 * @author <a href="mailto:jinzhongze@knet.cn">jinzhongze</a>
 * @version 1.0, 2011-11-11 16:26:09
 * @since 1.0
 */
public class DataCenterConsumer extends KnetMessageListener {

	public static final Logger logger = ShopLoggerFactory	.getLogger(DataCenterConsumer.class);

	public static  final ConcurrentMap<Class<? extends Domain>, SyncService> map = new ConcurrentHashMap<Class<? extends Domain>, SyncService>();
	
	@Resource
	private MqLogService mqLogService;
	
	//注册解析器
	public void setSyncMap(Map<Class<? extends Domain>, SyncService> syncMap) {
		map.putAll(syncMap);
		
		logger.warn("注册同步解析器...", map);
	}
	
	private int poolSize = 5;
	private ExecutorService worker = Executors.newFixedThreadPool(poolSize);  

	/*
	 * (non-Javadoc)
	 * @see cn.shangjia.common.mq.exec.KnetMessageListener#onKnetMessage(cn.shangjia.common.mq.exec.message.KnetMessage)
	 */
	@SuppressWarnings("unchecked")
	@Override
	public void onKnetMessage(final KnetMessage message) {
		
		if (message == null) 
			return;
		
		String messageId = message.getHead(KnetMessageHead.TransactionID);
		if (StringUtils.isEmpty(messageId))
			return;
		
		Collection<MessageBean> beans;
		try {
			beans = (Collection<MessageBean>) message.getBody("message");
		} catch (Exception e) {
			logger.warn("[同步系统], 消息体不是约定类型, body : ", message.getBody("message"));
			return;
		}
		
		if (beans == null || beans.isEmpty()) {
			logger.warn("[同步系统], 收到一条空消息, 消息ID : ", messageId);
			return ;
		}
		
		for (MessageBean bean : beans) {
			worker.execute(new ProcessTask(bean, message, mqLogService));
		}
		
	}
	
}
